#include <stdio.h>
#include <stdlib.h>
#include <signal.h>
#include <pthread.h>

#include "mytbf.h"

struct mytbf_ctx_st {
	int cps;
	int burst;
	int token;
	int pos;
	pthread_mutex_t mut;
	pthread_cond_t cond;
};

static pthread_t tid_alarm;
static pthread_once_t init_once = PTHREAD_ONCE_INIT;

static struct mytbf_ctx_st *job[MYTBF_MAX];
static pthread_mutex_t mut_job = PTHREAD_MUTEX_INITIALIZER;

static int get_free_pos_unlocked(void)
{
	int i;

	for (i=0;i<MYTBF_MAX;++i) {
		if (job[i]==NULL) {
			return i;
		}
	}
	return -1;
}

static int min(int a, int b)
{
	if (a<0 || b<0) {
		return 0;
	}
	if (a<b) {
		return a;
	}
	return b;
}

static void *thr_alarm(void *unused)
{
	int i;
	struct timespec ts;

	while (1) {
		ts.tv_sec = 1;
		ts.tv_nsec = 0;
		while (nanosleep(&ts, &ts));
		/* FIMME: 必须确定循环中有cancel point */
		for (i=0;i<MYTBF_MAX;++i) {
			if (job[i]) {
				pthread_mutex_lock(&job[i]->mut);
				job[i]->token += job[i]->cps;
				if (job[i]->token > job[i]->burst) {
					job[i]->token = job[i]->burst;
				}
				pthread_cond_broadcast(&job[i]->cond);
				pthread_mutex_unlock(&job[i]->mut);
			}
		}
	}
}

static void module_unload(void)
{
	int i;

	pthread_cancel(tid_alarm);
	pthread_join(tid_alarm, NULL);

	for (i=0;i<MYTBF_MAX;++i) {
		free(job[i]);
	}
}

static void module_load(void)
{
	int i, err;
	
	for (i=0;i<MYTBF_MAX;++i) {
		job[i]=NULL;
	}

	if (err=pthread_create(&tid_alarm, NULL, thr_alarm, NULL)) {
		fprintf(stderr, "ERROR: pthread_create(): %s\n", strerror(err));
		exit(1);
	}

	atexit(module_unload);
}

mytbf_ctx_t *mytbf_ctx_init(int cps, int burst)
{
	int pos;
	struct mytbf_ctx_st *node;

	pthread_once(&init_once, module_load);

	node = malloc(sizeof(*node));
	if (node==NULL) {
		perror("malloc()");
		return NULL;
	}

	node->cps = cps;
	node->burst = burst;
	node->token = 0;
	pthread_mutex_init(&node->mut, NULL);
	pthread_cond_init(&node->cond, NULL);

	// LOCK
	pthread_mutex_lock(&mut_job);

	if ((pos=get_free_pos_unlocked())<0) {
		//UNLOCK
		pthread_mutex_unlock(&mut_job);
		free(node);
		return NULL;
	}

	node->pos = pos;
	job[pos] = node;

	//UNLOCK
	pthread_mutex_unlock(&mut_job);

	return node;

	pthread_mutex_destroy(&node->mut);
	pthread_cond_destroy(&node->cond);

	free(node);

	return NULL;
}

int mytbf_ctx_destroy(mytbf_ctx_t *p)
{
	struct mytbf_ctx_st *node = p;

	pthread_mutex_lock(&mut_job);
	job[node->pos] = NULL;
	pthread_mutex_unlock(&mut_job);

	free(node);

	return 0;
}

ssize_t mytbf_request(mytbf_ctx_t *ctx, size_t num)
{
	ssize_t ret;
	struct mytbf_ctx_st *node = ctx;

	pthread_mutex_lock(&node->mut);
	while (node->token<=0) {
		pthread_cond_wait(&node->cond, &node->mut);
	}

	ret = min(node->token, num);
	node->token -= ret;

	pthread_mutex_unlock(&node->mut);

	return ret;
}

